package com.atguigu.flink.chapter5.transform;

import com.atguigu.flink.WaterSensor;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/19 14:56
 */
public class Flink12_Transform_Process {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        KeyedStream<WaterSensor, String> streamKeyed = env
          .socketTextStream("hadoop162", 9999)
          .map(line -> {
              String[] split = line.split(",");
              return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
          })
          .keyBy(WaterSensor::getId);
        // 实现1: sum
       /* streamKeyed
          .process(new KeyedProcessFunction<String, WaterSensor, Integer>() {
              private Map<String, Integer> sumMap = new HashMap<>();
              // 参数1: 输入的元素 参数2: 上下文对象, 存储了很多元数据信息 参数3: 把输出的元素放入这个收集器中
              @Override
              public void processElement(WaterSensor value, Context ctx, Collector<Integer> out) throws Exception {
                  String key = ctx.getCurrentKey();
                  System.out.println(key + "  " + value);
                  Integer sum = sumMap.get(key);
                  sumMap.put(key, sum == null ? value.getVc() : sum + value.getVc());
                  out.collect(sumMap.get(key));
              }
          })
          .print();*/
        // 实现2:map   sensor_1,20,20
       /* streamKeyed
          .process(new KeyedProcessFunction<String, WaterSensor, String>() {

              @Override
              public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                    out.collect(value.getId() + "," + value.getTs()+ "," + value.getVc());
              }
          })
          .print();*/


        env.execute();
    }
}
/*

 */
